/*
 * Licensed to the Deppon EXPRESS .
 */
package com.depponframework.rocketmq.spring.core;

import com.depponframework.rocketmq.spring.annotation.ConsumeMode;
import com.depponframework.rocketmq.spring.autoconfigure.RocketProperties;
import com.depponframework.rocketmq.spring.support.RocketUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.remoting.RPCHook;
import org.springframework.aop.framework.AopProxyUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.SmartLifecycle;
import org.springframework.util.Assert;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.Objects;

@Slf4j
public class RocketConsumerWrapper implements InitializingBean, DisposableBean, SmartLifecycle, ApplicationContextAware {

    private ApplicationContext applicationContext;
    private DefaultMQPushConsumer consumer;
    private boolean running;
    private RocketProperties.Consumer consumerProperties;
    private RocketListener rocketListener;
    private String consumerName;

    public RocketConsumerWrapper() {
    }

    private void setRunning(boolean running) {
        this.running = running;
    }

    public void setConsumerProperties(RocketProperties.Consumer consumerProperties) {
        this.consumerProperties = consumerProperties;
    }

    public void setRocketListener(RocketListener rocketListener) {
        this.rocketListener = rocketListener;
    }

    public void setConsumerName(String consumerName) {
        this.consumerName = consumerName;
    }

    private void initDefaultMQPushConsumer() throws MQClientException {
        Assert.notNull(this.rocketListener, "Property 'rocketListener' is required");
        Assert.notNull(this.consumerProperties.getConsumerGroup(), "Property 'consumerGroup' is required");
        Assert.notNull(this.consumerProperties.getNameServerAddress(), "Property 'nameServer' is required");
        Assert.notNull(this.consumerProperties.getTopic(), "Property 'topic' is required");

        RPCHook rpcHook = RocketUtils.getRPCHookByAkSk(this.applicationContext.getEnvironment(), this.consumerProperties.getAccessKey(), this.consumerProperties.getSecretKey());
        boolean enableMsgTrace = this.consumerProperties.getEnableMsgTrace();
        if (Objects.nonNull(rpcHook)) {
            this.consumer = new DefaultMQPushConsumer(
                    this.consumerProperties.getConsumerGroup(),
                    rpcHook,
                    new AllocateMessageQueueAveragely(),
                    enableMsgTrace,
                    this.applicationContext.getEnvironment().resolveRequiredPlaceholders(this.consumerProperties.getTopic()));
            this.consumer.setVipChannelEnabled(false);
            this.consumer.setInstanceName(RocketUtils.getInstanceName(rpcHook, this.consumerProperties.getConsumerGroup()));
        } else {
            log.debug("Access-key or secret-key not configure in " + this + ".");
            this.consumer = new DefaultMQPushConsumer(
                    this.consumerProperties.getConsumerGroup(),
                    enableMsgTrace,
                    this.applicationContext.getEnvironment().resolveRequiredPlaceholders(this.consumerProperties.getTopic()));
            this.consumer.setInstanceName(consumerName);
        }

        switch (this.consumerProperties.getMessageModel()) {
            case BROADCASTING:
                this.consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.BROADCASTING);
                break;
            case CLUSTERING:
                this.consumer.setMessageModel(org.apache.rocketmq.common.protocol.heartbeat.MessageModel.CLUSTERING);
                break;
            default:
                throw new IllegalArgumentException("Property 'messageModel' was wrong.");
        }

        consumer.setNamesrvAddr(this.consumerProperties.getNameServerAddress());
        consumer.subscribe(this.consumerProperties.getTopic(), String.join("||", this.consumerProperties.getTags()));
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.setPullBatchSize(this.consumerProperties.getPullBatchSize());
        consumer.setConsumeThreadMin(this.consumerProperties.getConsumeThreadMin());
        consumer.setConsumeThreadMax(this.consumerProperties.getConsumeThreadMax());
        consumer.setConsumeMessageBatchMaxSize(this.consumerProperties.getConsumeMessageBatchMaxSize());
        consumer.setPullInterval(this.consumerProperties.getPullInterval());
        consumer.setConsumeTimeout(this.consumerProperties.getConsumeTimeout());

        if (ConsumeMode.CONCURRENTLY.equals(this.consumerProperties.getConsumeMode())) {
            this.consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
                ConsumeConcurrentlyStatus consumeConcurrentlyStatus = rocketListener.consumeMessage(msgs, context);
                rocketListener.resultHandler(consumeConcurrentlyStatus);
                return consumeConcurrentlyStatus;
            });
        } else if (ConsumeMode.ORDERLY.equals(this.consumerProperties.getConsumeMode())) {
            this.consumer.registerMessageListener((MessageListenerOrderly) (msgs, context) -> {
                ConsumeOrderlyStatus consumeOrderlyStatus = rocketListener.consumeMessage(msgs, context);
                rocketListener.resultHandler(consumeOrderlyStatus);
                return consumeOrderlyStatus;
            });
        }
    }

    private Class getMessageType() {
        Class<?> targetClass = AopProxyUtils.ultimateTargetClass(this.rocketListener);
        Type[] interfaces = targetClass.getGenericInterfaces();

        for (Class superclass = targetClass.getSuperclass(); (Objects.isNull(interfaces) || 0 == interfaces.length) && Objects.nonNull(superclass); superclass = targetClass.getSuperclass()) {
            interfaces = superclass.getGenericInterfaces();
        }

        if (!Objects.nonNull(interfaces)) {
            return Object.class;
        }

        for (Type type : interfaces) {
            if (type instanceof ParameterizedType) {
                ParameterizedType parameterizedType = (ParameterizedType) type;
                if (Objects.equals(parameterizedType.getRawType(), RocketListener.class)) {
                    Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
                    if (Objects.nonNull(actualTypeArguments) && actualTypeArguments.length > 0) {
                        return (Class) actualTypeArguments[0];
                    }
                    return Object.class;
                }
            }
        }
        return Object.class;
    }

    //region super
    // TODO - InitializingBean : this method can be replaced with init-method
    @Override
    public void afterPropertiesSet() throws Exception {
        this.initDefaultMQPushConsumer();
        Class<?> messageType = this.getMessageType();
        log.debug("RocketMQ messageType: {}", messageType.getName());
    }

    // TODO - DisposableBean : this method can be replaced with destroy-method
    @Override
    public void destroy() {
        this.setRunning(false);
        if (Objects.nonNull(this.consumer)) {
            this.consumer.shutdown();
        }

        log.info("rocket mq consumer destroyed, {}", this.toString());
    }

    // TODO - Lifecycle : manage bean's lifecycle
    @Override
    public boolean isRunning() {
        return this.running;
    }

    @Override
    public void start() {
        if (this.running) {
            throw new IllegalStateException("consumer already running. " + this.toString());
        } else {
            try {
                consumer.start();
            } catch (MQClientException e) {
                throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
            }
            this.setRunning(true);
            log.debug("default rocket mq consumer for topic '{}' has been started!", this.consumerProperties.getTopic());
        }
    }

    @Override
    public void stop() {
        if (this.running) {
            if (Objects.nonNull(this.consumer)) {
                consumer.shutdown();
            }
            this.setRunning(false);
            log.debug("default rocket mq consumer for topic '{}' has been shutdown!", this.consumerProperties.getTopic());
        }
    }

    // TODO - ApplicationContextAware : this method is used to get spring context object
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
    //endregion
}
